package com.gallop.mq;

import com.gallop.mq.annotation.MQMessageListener;
import com.gallop.mq.core.*;
import com.gallop.mq.core.codec.NopMessageCodec;
import com.gallop.mq.core.dispatcher.MQMessageDispatcherImpl;
import com.gallop.mq.core.exception.MQException;
import com.gallop.mq.core.extend.NopExtendListenerScanner;
import com.gallop.mq.core.interceptor.DefaultNopInterceptor;
import com.gallop.mq.core.reliable.NopReliableDeliveryRegistry;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.ComponentScans;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.StringUtils;

import java.lang.reflect.Method;
import java.lang.reflect.Parameter;
import java.lang.reflect.ParameterizedType;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

@Slf4j
@Configuration
@EnableConfigurationProperties(MQProperties.class)
@ComponentScans({
        @ComponentScan("com.gallop.mq"),
})
public class MQAutoConfiguration implements SmartInitializingSingleton {

    @Autowired
    private MQProperties properties;
    @Autowired
    private ApplicationContext applicationContext;
    @Autowired
    private MQExecutorProvider executorProvider;
    @Autowired
    private MQReliableRegistry reliableRegistry;

    @Bean
    @ConditionalOnMissingBean(MQMessageCodec.class)
    @ConditionalOnProperty(prefix = "gallop.mq", value = "codec")
    public MQMessageCodec getCodec() {
        try {
            return properties.getCodec().getCodecClazz().newInstance();
        } catch (Exception e) {
            log.error("codec cant be created with class=" + properties.getCodec() + ", using nopCodec instead.", e);
        }
        return new NopMessageCodec();
    }

    @Bean(destroyMethod = "onDispose")
    @ConditionalOnMissingBean(MQTemplate.class)
    @ConditionalOnProperty(prefix = "gallop.mq", value = "hosts")
    public MQTemplate getTemplate() throws Exception {
        List<MQMessageListenerContainer> containers = listenerScan();
        ExtendListenerScanner scanner = applicationContext.getBean(ExtendListenerScanner.class);
        containers.addAll(scanner.provide());
        return templateInit(containers);
    }

    @Bean
    @ConditionalOnMissingBean(MQExecutorProvider.class)
    @ConditionalOnProperty(prefix = "gallop.mq", value = "hosts")
    public MQExecutorProvider getExecutorProvider() {
        int threadSize = Runtime.getRuntime().availableProcessors() * 2 + 1;
        return () -> new ThreadPoolExecutor(0, Math.max(5, threadSize),
                500L, TimeUnit.MILLISECONDS,
                new SynchronousQueue<>(), new MQExecutorFactory());
    }

    @Bean
    @ConditionalOnMissingBean(MQMessageDispatcher.class)
    @ConditionalOnProperty(prefix = "gallop.mq", value = "hosts")
    public MQMessageDispatcher getDispatcher() {
        return new MQMessageDispatcherImpl(reliableRegistry);
    }

    @Bean
    @ConditionalOnMissingBean(MQListenerInterceptor.class)
    @ConditionalOnProperty(prefix = "gallop.mq", value = "hosts")
    public MQListenerInterceptor getDefaultInterceptor() {
        return new DefaultNopInterceptor();
    }

    @Bean
    @ConditionalOnMissingBean(MQReliableRegistry.class)
    @ConditionalOnProperty(prefix = "gallop.mq", value = "hosts")
    public MQReliableRegistry getDefaultReliableRegistry() {
        return new NopReliableDeliveryRegistry();
    }

    @Bean
    @ConditionalOnMissingBean(ExtendListenerScanner.class)
    @ConditionalOnProperty(prefix = "gallop.mq", value = "hosts")
    public ExtendListenerScanner getDefaultScanner() {
        return new NopExtendListenerScanner();
    }

    private List<MQMessageListenerContainer> listenerScan() throws Exception {
        Method onMessageCallbackMethod = MQListener.class.getMethod("onMessage", Object.class);
        List<MQMessageListenerContainer> containers = Lists.newArrayList();

        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(MQMessageListener.class);
        beans.forEach((name, bean) -> {
            Class<?> beanType = AopUtils.isAopProxy(bean) ? AopUtils.getTargetClass(bean) : bean.getClass();
            if (MQListener.class.isAssignableFrom(beanType)) {
                // bean实现了MessageListener，注册onMessage方法
                MQMessageListener annotation = beanType.getAnnotation(MQMessageListener.class);
                MQMessageListenerContainer container =
                        createListenerContainer(name, bean, onMessageCallbackMethod, annotation, extractMessageTypeFromInterface(beanType));
                containers.add(container);
            }
            // 注册bean中所有包含MQMessageListener注解的方法
            Arrays.stream(beanType.getDeclaredMethods())
                    .filter(m -> m.isAnnotationPresent(MQMessageListener.class))
                    // onMessage equals: (name unequal + param count unequal)
                    .filter(m -> !(m.getName().equals(onMessageCallbackMethod.getName())
                            && m.getParameterCount() == onMessageCallbackMethod.getParameterCount()))
                    .peek(m -> m.setAccessible(true))
                    .map(m -> {
                        MQMessageListener annotation = m.getAnnotation(MQMessageListener.class);
                        return createListenerContainer(name + "#" + m.getName(), bean, m, annotation, extractMessageTypeFromMethod(m));
                    }).forEach(containers::add);
        });
        return containers;
    }

    private Class<?> extractMessageTypeFromInterface(Class<?> beanType) {
        return Arrays.stream(beanType.getGenericInterfaces())
                .filter(i -> ((ParameterizedType) i).getRawType().equals(MQListener.class))
                .findFirst()
                .map(i -> (Class<?>) ((ParameterizedType) i).getActualTypeArguments()[0])
                .orElse(null);
    }

    private Class<?> extractMessageTypeFromMethod(Method method) {
        return Arrays.stream(method.getParameters())
                .findFirst()
                .map(Parameter::getType)
                .orElse(null);
    }

    private MQMessageListenerContainer createListenerContainer(
            String name,
            Object bean,
            Method method,
            MQMessageListener annotation,
            Class<?> messageType) {
        if (messageType == null) {
            throw new MQException("MQMessageListener message parameter must be specified.name= " + name + " ,bean=" + bean);
        }
        if (!StringUtils.hasLength(annotation.topic())) {
            throw new MQException("MQMessageListener topic must be specified.name= " + name + " ,bean=" + bean);
        }
        MQMessageListenerContainer container = new MQMessageListenerContainer();
        container.setTopic(annotation.topic());
        container.setGroup(annotation.group());
        container.setInstance(bean);
        container.setMethod(method);
        container.setName(name);
        container.setMessageType(messageType);
        container.setInterceptor(Arrays.stream(annotation.interceptor())
                .map(type -> applicationContext.getBean(type))
                .collect(Collectors.toList()));
        return container;
    }

    private MQTemplate templateInit(List<MQMessageListenerContainer> containers) {
        MQTemplateInitializer initializer = getInitializer();
        MQMessageDispatcher dispatcher = getDispatcher();
        dispatcher.addRoutes(containers);
        return initializer.init(properties, dispatcher, getCodec(), executorProvider);
    }

    private MQTemplateInitializer getInitializer() {
        if (properties.getType() == SupportedType.CUSTOM) {
            try {
                return applicationContext.getBean(MQTemplateInitializer.class);
            } catch (Exception e) {
                log.error("", e);
            }
        }
        return properties.getType().getInitializer();
    }

    @Override
    public void afterSingletonsInstantiated() {
        log.info("gallop-mq components instantiated.");
    }

    private static class MQExecutorFactory implements ThreadFactory {

        private static final String threadName = "mq-client-pool";
        private final ThreadGroup threadGroup;
        private final AtomicInteger threadCount = new AtomicInteger();


        public MQExecutorFactory() {
            this.threadGroup = new ThreadGroup(threadName);
            threadGroup.setDaemon(true);
            threadGroup.setMaxPriority(Thread.NORM_PRIORITY);
        }

        @Override
        public Thread newThread(Runnable r) {
            return new Thread(threadGroup, r, threadName + "-worker-" + threadCount.getAndIncrement());
        }
    }
}
